package com.zhao.sparkstreaming.index_project

import java.sql.{Connection, PreparedStatement, ResultSet}

import org.apache.kafka.common.TopicPartition

import scala.collection.mutable

/**
 * Description: <br/>
 * Copyright (c) ，2020 ， 赵 <br/>
 * This program is protected by copyright laws. <br/>
 * Date： 2020/9/20 15:03
 *
 * @author 柒柒
 * @version : 1.0
 */

object OffsetUtils {
  def queryHistoryOffsetFromMySQL(appId: String, groupId: String): Map[TopicPartition, Long] = {
    val offsetMap = new mutable.HashMap[TopicPartition, Long]()
    //查询MySQL
    var connection: Connection = null
    var statement: PreparedStatement = null
    var resultSet: ResultSet = null
    try {
      connection = DruidConnectionPool.getConnection
      statement = connection.prepareStatement("SELECT topic_partition, offset FROM t_kafka_offset WHERE app_gid = ?")
      statement.setString(1, appId + "_" + groupId)
      val resultSet: ResultSet = statement.executeQuery()
      //变量结果集
      while (resultSet.next()) {
        val topic_partition = resultSet.getString(1)
        val offset = resultSet.getLong(2)
        val fields = topic_partition.split("_")
        val topic = fields(0)
        val partition = fields(1).toInt
        val topicPartition = new TopicPartition(topic, partition)
        offsetMap(topicPartition) = offset
      }
    } catch {
      case e: Exception => {
        throw new RuntimeException("查询历史偏移量出现异常")
      }
    } finally {
      if (resultSet != null) {
        resultSet.close()
      }
      if (statement != null) {
        statement.close()
      }
      if (connection != null) {
        connection.close()
      }
    }
    offsetMap.toMap
  }
}
